package com.diao.flink.windowAndTimer;

import org.apache.commons.lang.time.FastDateFormat;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

/**
 * @author: chenzhidiao
 * @date: 2021/2/3 10:57
 * @description: 通过自定义数据源来模拟消息的乱序问题，本来在启动后13秒发送的两条数据，其中有1条延迟了，到第19秒才进入系统
 * @version: 1.0
 *
 * 结果：
 * 当前时间：17:03:50
 * 当前系统时间：17:04:05
 * 窗口处理时间：17:04:05
 * 窗口开始时间：17:03:55
 * 窗口结束时间：17:04:05
 * =====================================================
 * (flink,1)
 * 当前系统时间：17:04:10
 * 窗口处理时间：17:04:10
 * 窗口开始时间：17:04:00
 * 窗口结束时间：17:04:10
 * =====================================================
 * (flink,3)
 * 当前系统时间：17:04:15
 * 窗口处理时间：17:04:15
 * 窗口开始时间：17:04:05
 * 窗口结束时间：17:04:15
 * =====================================================
 * (flink,2)
 *
 * 结果1，3，2 和正常的 2，3，1 不一致
 */
public class WindowWordCountWithDelayed {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);

        DataStreamSource<String> streamSource = env.addSource(new TestSourceDelayed());

        streamSource.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {

            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] words = s.split(",");
                for (String word : words) {
                    collector.collect(Tuple2.of(word,1));
                }
            }
        })
                .keyBy(tuple ->tuple.f0)
                .window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5)))
                .process(new SumProcessWindowFunction())
                .print().setParallelism(1);

        env.execute("wordCountWithWindow");


    }
}

class TestSourceDelayed implements SourceFunction<String> {



    FastDateFormat dateformat = FastDateFormat.getInstance("HH:mm:ss");

    @Override
    public void run(SourceContext<String> sourceContext) throws Exception {
        String currentTime = String.valueOf(System.currentTimeMillis());

        System.out.println(currentTime.substring(currentTime.length()-4));

        while(Integer.valueOf(currentTime.substring(currentTime.length()-4))>100){
            currentTime = String.valueOf(System.currentTimeMillis());
            continue;

        }
        System.out.println("当前时间："+dateformat.format(System.currentTimeMillis()));

        TimeUnit.SECONDS.sleep(13);

        String log = "flink";
        sourceContext.collect(log);

        TimeUnit.SECONDS.sleep(3);
        sourceContext.collect(log);

        TimeUnit.SECONDS.sleep(3);
        sourceContext.collect(log);

        TimeUnit.SECONDS.sleep(3000000);

    }

    @Override
    public void cancel() {

    }
}

